package com.urfresh.sp.flume.receive.configure;

import com.mongodb.MongoClient;
import com.mongodb.MongoClientOptions;
import com.mongodb.MongoCredential;
import com.mongodb.ServerAddress;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.SimpleMongoDbFactory;

import java.text.SimpleDateFormat;
import java.util.Arrays;
import java.util.Properties;

@Configuration
public class Config {

    Logger logger = LoggerFactory.getLogger(Config.class);

    @Value("${kafka.url}")
    String kafkaUrl;

    @Value("${kafka.group.id}")
    String kafkaGroupId;

    @Value("${kafka.timeout}")
    String kafkaTimeout;

    @Value("${kafka.auto.commit.interval}")
    String kafkaInterval;

    @Value("${mongoUrl}")
    String mongoUrl;
    @Value("${mongoPort}")
    int mongoPort;
    @Value("${mongoAuthName}")
    private String mongoAuthName;
    @Value("${mongoDbName}")
    private String mongoDbName;
    @Value("${mongoAuthPassword}")
    private String mongoAuthPassword;

    @Value("${mongoReplicaSet}")
    private String mongoReplicaSet;
    @Value("${mongoReadPreference}")
    private String mongoReadPreference;
    @Bean
    public SimpleDateFormat simpleDateFormat() {
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        return simpleDateFormat;
    }

    @Bean(name = "implantPoint")
    public KafkaConsumer sysConsumer() {
        return getKafkaConsumer();
    }

    @Bean(name = "appEventTrack")
    public KafkaConsumer appEventTrack() {
        return getKafkaConsumer();
    }

    @Bean(name = "pageTrack")
    public KafkaConsumer pageTrack() {
        return getKafkaConsumer();
    }

    private KafkaConsumer getKafkaConsumer() {
        Properties props = new Properties();
        props.put("bootstrap.servers", kafkaUrl);
        props.put("group.id", "urfresh");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", kafkaInterval);
        props.put("session.timeout.ms", kafkaTimeout);
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);

        return consumer;
    }

//    @Bean
//    public JedisPool jedisPool(){
//        JedisPoolConfig config = new JedisPoolConfig();
//        config.setMaxTotal(200);
//        config.setMaxIdle(50);
//        config.setMinIdle(8);//设置最小空闲数
//        config.setMaxWaitMillis(10000);
//        config.setTestOnBorrow(true);
//        config.setTestOnReturn(true);
//        //Idle时进行连接扫描
//        config.setTestWhileIdle(true);
//        //表示idle object evitor两次扫描之间要sleep的毫秒数
//        config.setTimeBetweenEvictionRunsMillis(30000);
//        //表示idle object evitor每次扫描的最多的对象数
//        config.setNumTestsPerEvictionRun(10);
//        //表示一个对象至少停留在idle状态的最短时间，然后才能被idle object evitor扫描并驱逐；这一项只有在timeBetweenEvictionRunsMillis大于0时才有意义
//        config.setMinEvictableIdleTimeMillis(60000);
//        return new JedisPool(config, redisHost, 6379, 0, redisPassword, dataBaseIndex);
//    }

    /*
	 * Factory bean that creates the com.mongodb.Mongo instance
	 */
    @Bean
    public MongoClient mongClient() throws Exception {
        MongoClient client = new MongoClient(
                Arrays.asList(
                        new ServerAddress(mongoUrl, mongoPort)),
                Arrays.asList(
                        MongoCredential.createCredential(mongoAuthName, mongoDbName, mongoAuthPassword.toCharArray())
                ),
                MongoClientOptions.builder()
                        .socketKeepAlive(true)  //是否保持长链接
                        .connectionsPerHost(200) // 最大连接数
                        .minConnectionsPerHost(20)// 最小连接数
//                        .threadsAllowedToBlockForConnectionMultiplier(10)
//                        .connectTimeout(5000)
//                        .maxWaitTime(5000)
//                        .requiredReplicaSetName(mongoReplicaSet)
//                        .readPreference(ReadPreference.secondaryPreferred())
//                        .socketTimeout(1000)
                        .build());
        return client;
    }

    @Bean
    public SimpleMongoDbFactory mongoFactory() throws Exception {
        return new SimpleMongoDbFactory(mongClient(), mongoDbName);
    }

    @Bean
    public MongoTemplate mongoTemplate() throws Exception {
        return new MongoTemplate(mongoFactory());
    }


}
